/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server;

import org.apache.jute.InputArchive;
import org.apache.jute.OutputArchive;
import org.apache.jute.Record;
import org.apache.log4j.Logger;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.proto.PrepareDataRequest;
import org.apache.zookeeper.server.paxos2.Storage;
import org.apache.zookeeper.txn.ErrorTxn;
import org.apache.zookeeper.txn.PrepareDataTxn;
import org.apache.zookeeper.txn.SetDataTxn;
import org.apache.zookeeper.txn.TxnHeader;
import org.apache.zookeeper.server.paxos2.*;

import java.util.*;
import java.io.IOException;

/**
 * This class maintains the tree data structure. It doesn't have any networking
 * or client connection code in it so that it can be tested in a stand alone
 * way.
 * <p/>
 * The tree maintains two parallel data structures: a hashtable that maps from
 * full paths to DataNodes and a tree of DataNodes. All accesses to a path is
 * through the hashtable. The tree is traversed only when serializing to disk.
 */
public class DataTree {
    private static final Logger LOG = Logger.getLogger(DataTree.class);

    private static Storage paxosStorage = null;
    private static boolean firstLoading = true;

    public DataTree() {

        List<Integer> stripes = Mapping.listShards(Configuration.serverId);
        if (paxosStorage == null)
            paxosStorage = new Storage(new ArrayList<Integer>(stripes), this);

    }

    public void setData(long[] blockId, byte[][] data, BlockMeta[] zxid, long time)
            throws KeeperException.NoNodeException {
        if (Configuration.debug) {
            for (int i = 0; i < blockId.length; i++)
                System.out.println("Write " + blockId[i] + " zxid=" + zxid[i]+" at "+System.currentTimeMillis());
        }

        paxosStorage.write(blockId, data, zxid);
    }

    public byte[][] getData(long[] blockId) throws KeeperException.NoNodeException {
        if (Configuration.debug) {
            for (int i = 0; i < blockId.length; i++)
                System.out.println("Read " + blockId[i]);
        }
        long start = System.currentTimeMillis();
        byte[][] data = new byte[blockId.length][];
        for (int i = 0; i < data.length; i++)
            data[i] = MemoryBuffer.allocateBuffer();
        //System.out.println("allocate time=" + (System.currentTimeMillis() - start));
        boolean[] valid = paxosStorage.read(blockId, data);
        for (int i = 0; i < valid.length; i++)
            if (!valid[i])
                data[i] = null;
        return data;
        //return new byte[1];
        // return paxosStorage.read(blockId);
        /*
           * DataNode n = nodes.get(path); if (n == null) { throw new
           * KeeperException.NoNodeException(); } synchronized (n) {
           * n.copyStat(stat); if (watcher != null) { dataWatches.addWatch(path,
           * watcher); } return n.data; }
           */
    }


    static public class ProcessTxnResult {
        public long clientId;

        public int cxid;

        public long zxid;

        public int err;

        public int type;

        public long blockId;

        /**
         * Equality is defined as the clientId and the cxid being the same. This
         * allows us to use hash tables to track completion of transactions.
         *
         * @see java.lang.Object#equals(java.lang.Object)
         */
        @Override
        public boolean equals(Object o) {
            if (o instanceof ProcessTxnResult) {
                ProcessTxnResult other = (ProcessTxnResult) o;
                return other.clientId == clientId && other.cxid == cxid;
            }
            return false;
        }

        /**
         * See equals() to find the rational for how this hashcode is generated.
         *
         * @see ProcessTxnResult#equals(Object)
         * @see java.lang.Object#hashCode()
         */
        @Override
        public int hashCode() {
            return (int) ((clientId ^ cxid) % Integer.MAX_VALUE);
        }

    }

    public volatile long lastProcessedZxid = 0;

    public ProcessTxnResult[] processTxn(TxnHeader header[], Record[] txn) {
        ProcessTxnResult[] rc = new ProcessTxnResult[header.length];
        for (int i = 0; i < rc.length; i++)
            rc[i] = new ProcessTxnResult();

        String debug = "";
        try {
            for (int i = 0; i < rc.length; i++) {
                rc[i].clientId = header[i].getClientId();
                rc[i].cxid = header[i].getCxid();
                rc[i].zxid = header[i].getZxid();
                rc[i].type = header[i].getType();
                rc[i].err = 0;

                if (Configuration.debug)
                    System.out.println("Processing " + rc[i].clientId + " " + rc[i].cxid + " " + header[i].getType());
                /*if (rc[i].type != OpCode.prepareData && rc[i].cxid != -1 && rc[i].cxid != ReplyCache.getLastCxid(rc[i].clientId) + 1 + i) {
                    if (rc[i].cxid < ReplyCache.getLastCxid(rc[i].clientId) + 1) {
                        System.out.println(rc.clientId + " expected " + (ReplyCache.getLastCxid(rc.clientId) + 1) + " but got " + rc.cxid);
                        return rc;
                    } else
                        throw new RuntimeException(rc.clientId + " expected " + (ReplyCache.getLastCxid(rc.clientId) + 1) + " but got " + rc.cxid);
                } */
            }
/*            if (rc[rc.length-1].zxid > lastProcessedZxid) {
                lastProcessedZxid = rc[rc.length-1].zxid;
            }*/


            switch (header[0].getType()) {
                case OpCode.setData:
                    long[] blockNo = new long[header.length];
                    byte[][] data = new byte[header.length][];
                    BlockMeta[] blockMetas = new BlockMeta[header.length];
                    //long[] zxid = new long[header.length];

                    for (int i = 0; i < header.length; i++) {
                        SetDataTxn setDataTxn = (SetDataTxn) txn[i];
                        debug = "Set data for  transaction for "
                                + setDataTxn.getSessionId() + " " + setDataTxn.getBlockId();
                        PrepareDataRequest p = PrepareDataCache.get(setDataTxn.getSessionId(), header[i].getCxid());
                        blockNo[i] = setDataTxn.getBlockId();
                        if (p != null)
                            data[i] = p.getData();
                        else
                            rc[i].err = -3;
                        blockMetas[i] = new BlockMeta(header[i].getZxid(),
                                (int) header[i].getClientId(), header[i].getCxid());

                    }
                    setData(blockNo, data,
                            blockMetas, header[0].getTime());
                    break;
                case OpCode.error:
                    for (int i = 0; i < txn.length; i++) {
                        ErrorTxn errTxn = (ErrorTxn) txn[i];
                        rc[i].err = errTxn.getErr();
                    }
                    break;
                case OpCode.prepareData:
                    for (int i = 0; i < txn.length; i++) {
                        PrepareDataTxn prepare = (PrepareDataTxn) txn[i];
                        PrepareDataCache.put(prepare.getSessionId(), prepare.getXid(), new PrepareDataRequest(prepare.getSessionId(), 0, prepare.getData()));
                    }
                    break;
                default:
                    throw new RuntimeException("Unknown txn type " + header[0].getType());
            }
            if (rc[rc.length - 1].zxid > lastProcessedZxid) {
                lastProcessedZxid = rc[rc.length - 1].zxid;
            }

        } catch (KeeperException e) {
            // These are expected errors since we take a lazy snapshot
            if (initialized
                    || (e.code() != Code.NONODE && e.code() != Code.NODEEXISTS)) {
                LOG.warn("Failed:" + debug, e);
            }
        }
        return rc;
    }


    int scount;

    public boolean initialized = false;


    public void serialize(OutputArchive oa, String tag) throws IOException {
        System.out.println("serialize request cache");
        PrepareDataCache.serialize(oa, "RequestCache");
        System.out.println("serialize reply cache");
        ReplyCache.serialize(oa, "ReplyCache");
        System.out.println("serialize storage");
        paxosStorage.serialize(oa, "storage");
        System.out.println("end serialize");
    }

    public void deserialize(InputArchive ia, String tag) throws IOException {
        PrepareDataCache.deserialize(ia, "RequestCache");
        ReplyCache.deserialize(ia, "ReplyCache");
        paxosStorage.deserialize(ia, "storage");
    }


    public void clear() {

    }

    public void syncWithOthers() {
        paxosStorage.syncWithOthers(this.lastProcessedZxid);
    }

}
